package org.codehaus.activemq.service.impl;

import EDU.oswego.cs.dl.util.concurrent.SynchronizedBoolean;
import org.codehaus.activemq.broker.BrokerClient;
import org.codehaus.activemq.service.Dispatcher;
import org.codehaus.activemq.service.MessageContainerManager;
import org.codehaus.activemq.service.Subscription;

public class DispatcherImpl
  implements Dispatcher
{
  private SynchronizedBoolean started = new SynchronizedBoolean(false);
  private DispatchWorker worker = new DispatchWorker();
  private MessageContainerManager containerManager;
  private Thread runner;

  public void register(MessageContainerManager mcm)
  {
    this.containerManager = mcm;
    this.worker.register(mcm);
  }

  public void wakeup(Subscription sub)
  {
    this.worker.wakeup();
  }

  public void wakeup()
  {
    this.worker.wakeup();
  }

  public void addActiveSubscription(BrokerClient client, Subscription sub)
  {
    this.worker.addActiveSubscription(client, sub);
  }

  public void removeActiveSubscription(BrokerClient client, Subscription sub)
  {
    this.worker.removeActiveSubscription(client, sub);
  }

  public void start()
  {
    if (this.started.commit(false, true)) {
      this.worker.start();
      this.runner = new Thread(this.worker);
      this.runner.setDaemon(true);
      this.runner.setPriority(6);
      this.runner.start();
    }
  }

  public void stop()
  {
    this.worker.stop();
    this.started.set(false);
  }
}